//
// Created by benxb on 2021/12/11.
//

#include "mem/log_manager.h"
#include <common/exception.h>
#include <common/logger.h>
#include <thread>
#include <unistd.h>
#include <common/util/file_util.h>

namespace benxdb {

LogManager::LogManager(const std::string& table_name, lsn_t lsn) : lsn_(lsn) {
    log_file_ = GlobalConfig::LogFilePath(table_name) + std::to_string(lsn);
    log_io_.open(log_file_, std::ios::binary | std::ios::in | std::ios::out);
    // directory or file does not exist
    if (!log_io_.is_open()) {
        log_io_.clear();
        // create a new file
        log_io_.open(log_file_, std::ios::binary | std::ios::trunc | std::ios::out);
        log_io_.close();
        // reopen with original mode
        log_io_.open(log_file_, std::ios::binary | std::ios::in | std::ios::out);
        if (!log_io_.is_open()) {
            throw Exception("can't open db file");
        }
    }
}

LogManager::~LogManager() {
    log_io_.close();
}

bool LogManager::AppendLog(Tuple& tuple) {
    if (read_only_) {
        return false;
    }
    log_io_.seekp(0, std::ios_base::end);
    log_io_.write(tuple.GetData(), tuple.GetSize());
    log_io_.flush();
    return true;
}

void LogManager::DeleteLogFile() {
    log_io_.close();
    remove(log_file_.data());
}

void LogManager::ChangeReadOnly() {
    read_only_ = true;
    log_io_.seekp(0);
}

LogManager::LogManager() {

}

void LogManager::ReadAllLog(const std::shared_ptr<Table>& table, std::vector<Tuple>& tuples) {
    std::vector<TypeId> schemas;
    table->GetSchema(schemas);
    if (FileUtil::GetFileSize(log_file_) == 0) {
        return;
    }
    log_io_.seekg(0, std::ios_base::beg);
    while (!log_io_.eof()) {
        tuples.emplace_back(log_io_, table->GetBytesSize());
    }
}

bool LogManager::AppendManyLog(std::vector<Tuple> &tuples) {
    if (read_only_) {
        return false;
    }
    log_io_.seekp(0, std::ios_base::end);
    for (auto& tuple : tuples) {
        log_io_.write(tuple.GetData(), tuple.GetSize());
    }
    log_io_.flush();
    return true;
}

bool LogManager::AppendManyLog(std::vector<Tuple> &tuples, int begin, int end) {
    if (read_only_) {
        return false;
    }
    log_io_.seekp(0, std::ios_base::end);
    for (int k = begin; k <= end; k++) {
        auto& tuple = tuples[k];
        log_io_.write(tuple.GetData(), tuple.GetSize());
    }
    log_io_.flush();
    return true;
}

bool LogManager::AppendManyLog(PointCloud<PointXYZI> &point_cloud, int begin, int end) {
    if (read_only_) {
        return false;
    }
    log_io_.seekp(0, std::ios_base::end);
    auto all_points = point_cloud.AllPoints();
    for (int k = begin; k <= end; k++) {
        auto& point = all_points->at(k);
        log_io_.write(reinterpret_cast<const char *>(&point.x), sizeof(point.x));
        log_io_.write(reinterpret_cast<const char *>(&point.y), sizeof(point.y));
        log_io_.write(reinterpret_cast<const char *>(&point.z), sizeof(point.z));
        log_io_.write(reinterpret_cast<const char *>(&point.intensity), sizeof(point.intensity));
    }
    log_io_.flush();
    return true;
}

bool LogManager::AppendPointLog(const char *point, int size) {
    if (read_only_) {
        return false;
    }
    log_io_.seekp(0, std::ios_base::end);
    log_io_.write(point, size);
    log_io_.flush();
    return true;
}

bool LogManager::AppendManyLog(PointCloud<PointSTXYZ> &point_cloud, int begin, int end) {
    if (read_only_) {
        return false;
    }
    log_io_.seekp(0, std::ios_base::end);
    auto all_points = point_cloud.AllPoints();
    for (int k = begin; k <= end; k++) {
        auto& point = all_points->at(k);
        log_io_.write(reinterpret_cast<const char *>(&point.sweep), sizeof(point.sweep));
        log_io_.write(reinterpret_cast<const char *>(&point.timestamp), sizeof(point.timestamp));
        log_io_.write(reinterpret_cast<const char *>(&point.x), sizeof(point.x));
        log_io_.write(reinterpret_cast<const char *>(&point.y), sizeof(point.y));
        log_io_.write(reinterpret_cast<const char *>(&point.z), sizeof(point.z));
    }
    log_io_.flush();
    return true;
}

bool LogManager::AppendManyLog(PointCloud<PointSTXYZC> &point_cloud, int begin, int end) {
    if (read_only_) {
        return false;
    }
    log_io_.seekp(0, std::ios_base::end);
    auto all_points = point_cloud.AllPoints();
    for (int k = begin; k <= end; k++) {
        auto& point = all_points->at(k);
        log_io_.write(reinterpret_cast<const char *>(&point.sweep), sizeof(point.sweep));
        log_io_.write(reinterpret_cast<const char *>(&point.timestamp), sizeof(point.timestamp));
        log_io_.write(reinterpret_cast<const char *>(&point.classification), sizeof(point.classification));
        log_io_.write(reinterpret_cast<const char *>(&point.x), sizeof(point.x));
        log_io_.write(reinterpret_cast<const char *>(&point.y), sizeof(point.y));
        log_io_.write(reinterpret_cast<const char *>(&point.z), sizeof(point.z));
    }
    log_io_.flush();
    return true;
}

}